package com.examples.mysql2mysql.batch;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


public class MySQLToMySQL {

    /***
     *  批处理MySQL数据表
     * **/
    public static void main(String[] args) throws Exception {
        // 设置Flink的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 创建MySQL源输入
        DataStream<Tuple2<String, Integer>> input = env.addSource(new JdbcSourceFunction());

        // 进行转换等操作
        DataStream<Tuple2<String, Integer>> transformed = input.map(new MapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(Tuple2<String, Integer> value) throws Exception {
                String name = value.f0.toUpperCase();
                int del = value.f1 + 1;
                return new Tuple2<>(name, del);
            }
        }).returns(Types.<Tuple2<String, Integer>>TUPLE(Types.STRING, Types.INT)); // 添加 returns(...) 方法来指定输出类型

        // 将数据写入MySQL目标
        transformed.addSink(new JdbcSinkFunction());

        // 执行任务
        env.execute("MySQL to Flink to MySQL");
    }
}
